package com.navinfo.platform.fault.service.week;


import com.navinfo.platform.fault.dto.BasicDataDto;
import com.navinfo.platform.fault.service.day.BasicDataService;
import com.navinfo.platform.fault.utils.day.VehicleBasicDataSchema;
import com.navinfo.platform.fault.utils.week.VehicleBasicDataWeekUtil;
import com.navinfo.platform.common.service.ICommonService;
import com.navinfo.platform.common.service.IDataHandleService;
import com.navinfo.platform.common.utils.DateUtils;
import com.navinfo.platform.common.utils.StatisticsDef;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public enum BasicDataWeekHandleService implements IDataHandleService {

    INS;

    @Override
    public void handle(List<Object> obj, Map<String, String> configMap, SparkSession spark) {
        Dataset<Row> baiscDataset = castValue(obj.get(0));
        JavaRDD<Row> saveDataRDD = baiscDataset.toJavaRDD().mapPartitions(rowToBasicDataDto());
        String day= DateUtils.getStartEndDay(configMap.get(ICommonService.DAY_VALUE), StatisticsDef.WEEK)._1;
        StructType schema = VehicleBasicDataSchema.createSchema();
        Dataset<Row> saveDataset = spark.createDataFrame(saveDataRDD, schema);
        BasicDataService.saveData(spark,saveDataset, configMap, "hdfs.vehicle.week.save.tableName", day,schema);
    }


    @SuppressWarnings("unchecked")
    public static FlatMapFunction<Iterator<Row>, Row> rowToBasicDataDto() {
        return itor -> {
            List<Row> basicDataDtos = new ArrayList<>();
            StructType structType = VehicleBasicDataSchema.createSchema();
            while(itor.hasNext()){
                Row row = itor.next();
                BasicDataDto basicDataDto = VehicleBasicDataWeekUtil.parseRowToBasicDataDto(row);
                basicDataDtos.add(VehicleBasicDataSchema.createRow(basicDataDto, structType));
            }
            return basicDataDtos.iterator();
        };
    }


}
